package com.atguigu.chapter11;

import com.atguigu.bean.WaterSensor;
import org.apache.flink.api.common.functions.MapFunction;
import org.apache.flink.api.java.tuple.Tuple2;
import org.apache.flink.streaming.api.datastream.DataStreamSource;
import org.apache.flink.streaming.api.datastream.SingleOutputStreamOperator;
import org.apache.flink.streaming.api.environment.StreamExecutionEnvironment;
import org.apache.flink.table.api.Table;
import org.apache.flink.table.api.bridge.java.StreamTableEnvironment;
import org.apache.flink.types.Row;

import static org.apache.flink.table.api.Expressions.$;

/**
 * Author: Pepsi
 * Date: 2023/8/24
 * Desc:
 */
public class Flink03_Table_BaseUse {
    public static void main(String[] args) {

        // 获取流执行环境
        StreamExecutionEnvironment env = StreamExecutionEnvironment.getExecutionEnvironment();
        env.setParallelism(1);

        DataStreamSource<WaterSensor> stream = env.fromElements(
                new WaterSensor("sensor_1", 1000L, 10),
                new WaterSensor("sensor_1", 2000L, 20),
                new WaterSensor("sensor_2", 3000L, 30),
                new WaterSensor("sensor_1", 4000L, 40),
                new WaterSensor("sensor_1", 5000L, 50),
                new WaterSensor("sensor_1", 6000L, 60)
        );

        // 获取表执行环境，在这之前要获取流的执行环境作为参数传过来
        StreamTableEnvironment tEnv = StreamTableEnvironment.create(env);

        // 1. 用环境，将流转换成表
        Table table = tEnv.fromDataStream(stream);

        // 2. 对表对象进行查询
        // select id,sum(vc) from t group by id
        Table result = table
                .groupBy($("id"))
                .select($("id"),$("vc").sum().as("sum_vc"));  // 可以这样写，直接取的时候再聚合，简单


        // 表执行完直接输出，这种只能在测试的时候用
        result.execute().print();

    }
}
